package com.htiiot.hbase;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * 用于将tsdb表的file通过tid进行过滤生成SequenceFile
 */
@InterfaceAudience.Public
@InterfaceStability.Stable
public class TsdbRefact {
    private static final Log LOG = LogFactory.getLog(org.apache.hadoop.hbase.mapreduce.Export.class);
    final static String NAME = "RefactTableOf-";
    final static String RAW_SCAN = "hbase.mapreduce.include.deleted.rows";
    final static String EXPORT_BATCHING = "hbase.export.scanner.batch";
    final static String tmpDir = "/tmp/tidobj";

    static class ExportTableMapper
            extends TableMapper<ImmutableBytesWritable, Result> {
        private static List<byte[]> tidList;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            System.out.println("Running map setup function...");
            FileSystem fs = FileSystem.get(context.getConfiguration());
            //read tidList
            FSDataInputStream in = fs.open(new Path(tmpDir));
            ObjectInputStream oin = new ObjectInputStream(in);
            try {
                tidList = (List<byte[]>) oin.readObject();
                if(tidList!=null&&tidList.isEmpty()){
                    System.out.println("tid list is empty!!!");
                }else{
                    System.out.println("List size is: "+tidList.size());
                }
            }catch (ClassNotFoundException e){
                e.printStackTrace();
                throw new InterruptedIOException(e.getMessage());
            }finally {
                oin.close();
            }
        }

        @Override
        public void map(ImmutableBytesWritable key, Result value, Context context)
                throws IOException, InterruptedException {
            if(!judge(key.get())) {
                context.write(key, value);
            }
        }

        public boolean judge(byte[] rowkey){
            for(byte[] x:tidList){
                if(Arrays.equals(x,Arrays.copyOfRange(rowkey,14,17))){
                    return true;
                }
            }
            return false;
        }
    }

    public static Job createSubmittableJob(Configuration conf, String[] args)
            throws IOException {
        String tableName = "tsdb";
        Path outputDir = new Path(args[0]);
        Job job = new Job(conf, NAME+tableName);
        job.setJarByClass(TsdbRefact.class);
        // Set optional scan parameters
        Scan s = getConfiguredScanForJob(conf, args);
        IdentityTableMapper.initJob(tableName, s, ExportTableMapper.class, job);
        // No reducers.  Just write straight to output files.
        job.setNumReduceTasks(0);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        job.setOutputKeyClass(ImmutableBytesWritable.class);
        job.setOutputValueClass(Result.class);
        FileOutputFormat.setOutputPath(job, outputDir); // job conf doesn't contain the conf so doesn't have a default fs.
        return job;
    }

    //args[0]->outPutPath args[1]->tids
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            usage("Wrong number of arguments: " + otherArgs.length);
            System.exit(-1);
        }
        List<byte[]> tidList = new ArrayList<byte[]>();
        List<Get> getList = new ArrayList<Get>();
        for (String id:otherArgs[1].split(",")){
            getList.add(new Get(id.getBytes()));
        }
        Connection conn = ConnectionFactory.createConnection(conf);
        Table table = conn.getTable(TableName.valueOf("tsdb-uid"));
        for (Result r:table.get(getList)){
            Cell c = r.getColumnLatestCell("id".getBytes(), "tagv".getBytes());
            if(c!=null) {
                byte[] bytes = CellUtil.cloneValue(c);
                tidList.add(bytes);
                System.out.println(Arrays.toString(bytes));
            }
        }
        conn.close();
        FileSystem fs = FileSystem.get(conf);
        //write tid array to hdfs
        FSDataOutputStream outputStream = fs.create(new Path(tmpDir));
        ObjectOutputStream oos = new ObjectOutputStream(outputStream);
        oos.writeObject(tidList);
        oos.close();
        Job job = createSubmittableJob(conf, otherArgs);
        job.waitForCompletion(true);
        //clear tmpDir data
        fs.delete(new Path(tmpDir),true);
        fs.close();
    }

    private static Scan getConfiguredScanForJob(Configuration conf, String[] args) throws IOException {
        Scan s = new Scan();
        // Optional arguments.
        // Set Scan Versions
        int versions = args.length > 2? Integer.parseInt(args[2]): 1;
        s.setMaxVersions(versions);
        // Set Scan Range
        long startTime = args.length > 3? Long.parseLong(args[3]): 0L;
        long endTime = args.length > 4? Long.parseLong(args[4]): Long.MAX_VALUE;
        s.setTimeRange(startTime, endTime);
        // Set cache blocks
        s.setCacheBlocks(false);
        // set Start and Stop row
        if (conf.get(TableInputFormat.SCAN_ROW_START) != null) {
            s.setStartRow(Bytes.toBytes(conf.get(TableInputFormat.SCAN_ROW_START)));
        }
        if (conf.get(TableInputFormat.SCAN_ROW_STOP) != null) {
            s.setStopRow(Bytes.toBytes(conf.get(TableInputFormat.SCAN_ROW_STOP)));
        }
        // Set Scan Column Family
        boolean raw = Boolean.parseBoolean(conf.get(RAW_SCAN));
        if (raw) {
            s.setRaw(raw);
        }

        if (conf.get(TableInputFormat.SCAN_COLUMN_FAMILY) != null) {
            s.addFamily(Bytes.toBytes(conf.get(TableInputFormat.SCAN_COLUMN_FAMILY)));
        }
        // Set RowFilter or Prefix Filter if applicable.
        Filter exportFilter = getExportFilter(args);
        if (exportFilter!= null) {
            LOG.info("Setting Scan Filter for Export.");
            s.setFilter(exportFilter);
        }

        int batching = conf.getInt(EXPORT_BATCHING, -1);
        if (batching !=  -1){
            try {
                s.setBatch(batching);
            } catch (IncompatibleFilterException e) {
                LOG.error("Batching could not be set", e);
            }
        }
        LOG.info("versions=" + versions + ", starttime=" + startTime +
                ", endtime=" + endTime + ", keepDeletedCells=" + raw);
        return s;
    }

    private static Filter getExportFilter(String[] args) {
        Filter exportFilter = null;
        String filterCriteria = (args.length > 5) ? args[5]: null;
        if (filterCriteria == null) return null;
        if (filterCriteria.startsWith("^")) {
            String regexPattern = filterCriteria.substring(1, filterCriteria.length());
            exportFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(regexPattern));
        } else {
            exportFilter = new PrefixFilter(Bytes.toBytes(filterCriteria));
        }
        return exportFilter;
    }

    /*
     * @param errorMsg Error message.  Can be null.
     */
    private static void usage(final String errorMsg) {
        if (errorMsg != null && errorMsg.length() > 0) {
            System.err.println("ERROR: " + errorMsg);
        }
        System.err.println("Usage: Export [-D <property=value>]* <tablename> <outputdir> [<versions> " +
                "[<starttime> [<endtime>]] [^[regex pattern] or [Prefix] to filter]]\n");
        System.err.println("  Note: -D properties will be applied to the conf used. ");
        System.err.println("  For example: ");
        System.err.println("   -D mapreduce.output.fileoutputformat.compress=true");
        System.err.println("   -D mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.GzipCodec");
        System.err.println("   -D mapreduce.output.fileoutputformat.compress.type=BLOCK");
        System.err.println("  Additionally, the following SCAN properties can be specified");
        System.err.println("  to control/limit what is exported..");
        System.err.println("   -D " + TableInputFormat.SCAN_COLUMN_FAMILY + "=<familyName>");
        System.err.println("   -D " + RAW_SCAN + "=true");
        System.err.println("   -D " + TableInputFormat.SCAN_ROW_START + "=<ROWSTART>");
        System.err.println("   -D " + TableInputFormat.SCAN_ROW_STOP + "=<ROWSTOP>");
        System.err.println("For performance consider the following properties:\n"
                + "   -Dhbase.client.scanner.caching=100\n"
                + "   -Dmapreduce.map.speculative=false\n"
                + "   -Dmapreduce.reduce.speculative=false");
        System.err.println("For tables with very wide rows consider setting the batch size as below:\n"
                + "   -D" + EXPORT_BATCHING + "=10");
    }

}